package cn.minds.capture.dispatch.server;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import cn.minds.capture.dispatch.Connect;
import cn.minds.capture.dispatch.TaskManager;
import cn.minds.capture.dispatch.task.Task;
import cn.minds.capture.dispatch.task.TaskHandler;
import cn.minds.capture.dispatch.task.TaskHandlerFacotry;

public class WorkerManager implements TaskManager {

	private final static int DEFAULT_IO_THREADS = Runtime.getRuntime().availableProcessors() * 2;
	BlockingQueue<Task> taskQueue = new LinkedBlockingQueue<Task>(10);
	private volatile boolean stop = false;
	private int index = 0;
	private TaskHandlerFacotry taskHandlerFacotry;
	private Connect connect;

	public WorkerManager(TaskHandlerFacotry taskHandlerFacotry) {
		this.taskHandlerFacotry = taskHandlerFacotry;
		Worker[] workers = new Worker[DEFAULT_IO_THREADS];
		for (int i = 0; i < DEFAULT_IO_THREADS; i++) {
			workers[i] = new Worker(this, i);
		}
		for (Worker worker : workers) {
			worker.start();
		}
	}

	// 设置连接
	public void setConnect(Connect connect) {
		this.connect = connect;
	}

	// 向外发送一个Task
	public void sendTask(Task task) {
		connect.sendCommand(task);
	}

	// 从连接中得到一个Task
	public void receiveTask(Task task) {
		try {
			taskQueue.put(task);
			System.out.println(" :" + task.getId());
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	// 分配任务线程一个Task
	public Task distributeTask() {
		Task task = taskQueue.poll();
		return task;
	}

	// 是否启动成功
	public synchronized void waitNotify() {
		index++;
		while (index != DEFAULT_IO_THREADS) {
		}
	}

	// 是否已经停止
	public boolean isStop() {
		return stop;
	}

	// 创建用户操作的
	public TaskHandler createTaskHandler() {
		return taskHandlerFacotry.createTaskHandler(this);
	}

	// 创建一个工作组
	public final static WorkerManager createTaskHandler(TaskHandlerFacotry taskHandlerFacotry) {
		return new WorkerManager(taskHandlerFacotry);
	}

}
